package com.staryea.newcustomer.task;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.staryea.invocate.config.KafkaSender;
import com.staryea.invocate.constant.OperationConsts;
import com.staryea.newcustomer.module.User;
import com.staryea.newcustomer.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Auther: hs
 * @Date: 2019/3/4 10:53
 * @Description:
 */
@Component
@Slf4j
public class UserTask {

    private final KafkaSender kafkaSender;
    private final UserService userService;
    @Value("${kafka.listener.topics}")
    private String topic;

    public UserTask(KafkaSender kafkaSender, UserService userService) {
        this.kafkaSender = kafkaSender;
        this.userService = userService;
    }


    @Scheduled(fixedRate = 10 * 1000)
    public void addUserTask() {
        /*User user=new User();
        user.setUserName("HS");
        user.setDescription("text");
        user.setCreateTime(LocalDateTime.now());*/
        List<User> users = userService.listUsers();
        List<LocalDateTime> collect = users.stream().limit(10).distinct().map(e -> e.getCreateTime()).collect(Collectors.toList());
        String JSONUser = JSON.toJSONStringWithDateFormat(collect,
                OperationConsts.DateTimeFormat.DATE_TIME_PATTERN,
                SerializerFeature.PrettyFormat);
        kafkaSender.sendMessage(topic, JSONUser);
        log.info("collect: {}", JSONUser);
    }
}

